#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""
不安全的反序列化(Insecure Deserialization)漏洞扫描脚本
检测目标应用是否存在不安全的反序列化漏洞
"""

import requests
import time
import re
import urllib.parse
import base64
import json
import pickle
from typing import Dict, Any, List
import logging
from concurrent.futures import ThreadPoolExecutor, as_completed

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# 脚本元数据
NAME = "不安全的反序列化漏洞扫描器"
DESCRIPTION = "检测目标应用是否存在不安全的反序列化漏洞，支持多种序列化格式的测试"
AUTHOR = "Security Team"
VERSION = "1.0.0"
VULNERABILITY_TYPE = "insecure_deserialization"
SEVERITY = "critical"
ENABLED = True

# 必要参数
REQUIRED_PARAMS = ["target"]

# 可选参数及默认值
OPTIONAL_PARAMS = {
    "scan_depth": 2,
    "timeout": 10,
    "max_retries": 3,
    "test_params": [],  # 自定义测试参数
    "method": "get"  # get或post
}

# 不同语言/框架的序列化特征
SERIALIZATION_FORMATS = [
    {
        "name": "Python Pickle",
        "content_type": "application/octet-stream",
        "header": b'\x80',
        "test_payload": lambda: base64.b64encode(pickle.dumps({"test": "value"})).decode(),
        "payload_marker": "pickle_test"
    },
    {
        "name": "Java Serialization",
        "content_type": "application/octet-stream",
        "header": b'\xac\xed',
        "test_payload": lambda: "rO0ABXNyABFqYXZhLnV0aWwuSGFzaE1hcAUH2sHDFmDRAwACRgAKbG9hZEZhY3RvckkACXRocmVzaG9sZHhwP0AAAAAAAAx3CAAAABAAAAACdAAKdGVzdFZhbHVl",  # 简单的Java序列化对象base64编码
        "payload_marker": "testValue"
    },
    {
        "name": "PHP Serialization",
        "content_type": "text/plain",
        "header": b'a:',
        "test_payload": lambda: "YToxOntzOjQ6InRlc3QiO3M6NToidmFsdWUiO30=",  # a:1:{s:4:"test";s:5:"value";} base64编码
        "payload_marker": "test"
    },
    {
        "name": "JSON (可能存在对象注入)",
        "content_type": "application/json",
        "header": b'{',
        "test_payload": lambda: base64.b64encode(json.dumps({"test": "value"}).encode()).decode(),
        "payload_marker": "test"
    }
]

# 常见的反序列化参数名
COMMON_DESERIALIZATION_PARAMS = [
    "data", "payload", "serialized", "ser", "object", "obj",
    "user", "session", "cookie", "token", "auth", "login",
    "state", "context", "info", "profile", "settings", "config",
    "data1", "data2", "param", "params", "value", "values",
    "hash", "key", "secret", "cred", "credentials", "payload"
]

# 反序列化漏洞常见的错误特征
DESERIALIZATION_ERRORS = {
    "python": [
        "UnpicklingError",
        "pickle",
        "AttributeError",
        "ImportError",
        "ModuleNotFoundError",
        "invalid load key"
    ],
    "java": [
        "java.io.IOException",
        "ClassNotFoundException",
        "InvalidClassException",
        "StreamCorruptedException",
        "NotSerializableException"
    ],
    "php": [
        "unserialize()",
        "Notice: unserialize()",
        "Warning: unserialize()",
        "PHP Warning",
        "PHP Notice"
    ],
    "general": [
        "deserializ",
        "serializ",
        "object reference",
        "corrupt data",
        "invalid data",
        "parse error",
        "format error"
    ]
}

def detect_deserialization_error(response_text: str) -> Dict[str, Any]:
    """
    分析响应文本，检测是否存在反序列化错误
    """
    indicators = {
        "error_detected": False,
        "language": None,
        "framework": None,
        "indicators": []
    }
    
    # 检查Python错误
    for indicator in DESERIALIZATION_ERRORS["python"]:
        if indicator in response_text:
            indicators["error_detected"] = True
            indicators["language"] = "python"
            indicators["indicators"].append(indicator)
            break
    
    # 检查Java错误
    if not indicators["error_detected"]:
        for indicator in DESERIALIZATION_ERRORS["java"]:
            if indicator in response_text:
                indicators["error_detected"] = True
                indicators["language"] = "java"
                indicators["indicators"].append(indicator)
                break
    
    # 检查PHP错误
    if not indicators["error_detected"]:
        for indicator in DESERIALIZATION_ERRORS["php"]:
            if indicator in response_text:
                indicators["error_detected"] = True
                indicators["language"] = "php"
                indicators["indicators"].append(indicator)
                break
    
    # 检查一般错误
    if not indicators["error_detected"]:
        for indicator in DESERIALIZATION_ERRORS["general"]:
            if indicator.lower() in response_text.lower():
                indicators["error_detected"] = True
                indicators["indicators"].append(indicator)
    
    # 检查是否有服务器错误
    if "500 Internal Server Error" in response_text or "Server Error" in response_text:
        indicators["server_error"] = True
    
    return indicators

def check_base64_encoded_parameter(target: str, param_name: str, method: str, timeout: int) -> Dict[str, Any]:
    """
    检查参数是否使用base64编码
    """
    result = {
        "base64_encoded": False,
        "param": param_name,
        "decoded_value": None,
        "encoding_type": None,
        "error": None
    }
    
    try:
        # 发送一个正常请求，获取参数值进行分析
        if method.lower() == "get":
            parsed_url = urllib.parse.urlparse(target)
            params = dict(urllib.parse.parse_qsl(parsed_url.query))
            
            if param_name in params:
                param_value = params[param_name]
                
                # 检查是否是base64编码
                try:
                    # 确保参数长度是4的倍数（base64要求）
                    padded_value = param_value + '=' * ((4 - len(param_value) % 4) % 4)
                    decoded_bytes = base64.b64decode(padded_value, validate=True)
                    decoded_value = decoded_bytes.decode('utf-8', errors='ignore')
                    
                    # 检查解码后的值是否有意义
                    if len(decoded_value) > 0 and any(c.isalnum() for c in decoded_value):
                        result["base64_encoded"] = True
                        result["decoded_value"] = decoded_value
                        
                        # 尝试识别序列化格式
                        if decoded_value.startswith('{') and decoded_value.endswith('}'):
                            result["encoding_type"] = "json"
                        elif decoded_value.startswith('a:') or decoded_value.startswith('O:'):
                            result["encoding_type"] = "php_serialized"
                except Exception:
                    pass
    
    except Exception as e:
        result["error"] = str(e)
    
    return result

def test_parameter_for_deserialization(target: str, param_name: str, serialization_format: dict, method: str, timeout: int) -> Dict[str, Any]:
    """
    测试特定参数是否存在不安全的反序列化漏洞
    """
    result = {
        "vulnerable": False,
        "param": param_name,
        "serialization_format": serialization_format["name"],
        "status_code": None,
        "response_time": None,
        "error": None,
        "error_indicators": []
    }
    
    try:
        # 准备请求头和数据
        headers = {
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
            "Accept": "*/*",
        }
        
        # 获取测试payload
        test_payload = serialization_format["test_payload"]()
        
        # 记录开始时间
        start_time = time.time()
        
        # 根据请求方法执行请求
        if method.lower() == "get":
            # 构建URL参数
            params = {param_name: test_payload}
            
            # 发送请求
            response = requests.get(
                target,
                params=params,
                headers=headers,
                timeout=timeout,
                verify=False,
                allow_redirects=True
            )
        else:  # post
            # 准备数据
            data = {param_name: test_payload}
            
            # 添加一些常见的其他参数
            for common_param in COMMON_DESERIALIZATION_PARAMS[:5]:
                if common_param != param_name:
                    data[common_param] = "test"
            
            # 设置内容类型
            headers["Content-Type"] = serialization_format["content_type"]
            
            # 发送请求
            response = requests.post(
                target,
                data=data,
                headers=headers,
                timeout=timeout,
                verify=False,
                allow_redirects=True
            )
        
        # 记录响应时间
        end_time = time.time()
        result["response_time"] = end_time - start_time
        result["status_code"] = response.status_code
        
        # 检查响应是否包含反序列化错误
        error_indicators = detect_deserialization_error(response.text)
        if error_indicators["error_detected"]:
            result["vulnerable"] = True
            result["error_indicators"] = error_indicators["indicators"]
            result["language"] = error_indicators.get("language", "unknown")
        
        # 检查是否有服务器错误
        if response.status_code == 500 or error_indicators.get("server_error", False):
            result["server_error"] = True
            # 服务器错误可能表明存在反序列化漏洞
            if not result["vulnerable"]:
                result["suspicious"] = True
                result["error_indicators"].append("server_error_500")
        
        # 检查响应中是否包含payload标记
        if serialization_format["payload_marker"] in response.text:
            result["payload_reflected"] = True
    
    except requests.exceptions.Timeout:
        result["error"] = "Request timeout"
    except requests.exceptions.RequestException as e:
        result["error"] = f"Request error: {str(e)}"
    except Exception as e:
        result["error"] = f"Unexpected error: {str(e)}"
    
    return result

def test_url_parameters(target: str, serialization_format: dict, method: str, timeout: int) -> Dict[str, Any]:
    """
    测试URL中所有参数是否存在不安全的反序列化漏洞
    """
    # 解析URL获取现有参数
    parsed_url = urllib.parse.urlparse(target)
    existing_params = dict(urllib.parse.parse_qsl(parsed_url.query))
    
    result = {
        "vulnerable": False,
        "existing_params": list(existing_params.keys()),
        "base64_encoded_params": [],
        "results": []
    }
    
    # 测试所有现有参数
    for param_name in existing_params.keys():
        # 首先检查是否是base64编码
        base64_check = check_base64_encoded_parameter(target, param_name, method, timeout)
        if base64_check["base64_encoded"]:
            result["base64_encoded_params"].append({
                "param": param_name,
                "decoded_value": base64_check["decoded_value"],
                "encoding_type": base64_check["encoding_type"]
            })
        
        # 测试反序列化
        param_result = test_parameter_for_deserialization(
            target, param_name, serialization_format, method, timeout
        )
        result["results"].append(param_result)
        
        if param_result["vulnerable"]:
            result["vulnerable"] = True
    
    return result

def test_common_parameters(target: str, serialization_format: dict, method: str, timeout: int) -> Dict[str, Any]:
    """
    测试常见参数名是否存在不安全的反序列化漏洞
    """
    # 测试前15个常见参数
    param_names = COMMON_DESERIALIZATION_PARAMS[:15]
    
    result = {
        "vulnerable": False,
        "tested_params": param_names,
        "results": []
    }
    
    # 测试每个参数
    for param_name in param_names:
        param_result = test_parameter_for_deserialization(
            target, param_name, serialization_format, method, timeout
        )
        result["results"].append(param_result)
        
        if param_result["vulnerable"]:
            result["vulnerable"] = True
    
    return result

def scan(target: str, **kwargs) -> Dict[str, Any]:
    """
    执行不安全的反序列化漏洞扫描
    
    Args:
        target: 目标URL
        **kwargs: 可选参数
            scan_depth: 扫描深度 (1-5)
            timeout: 请求超时时间
            max_retries: 最大重试次数
            test_params: 自定义测试参数列表
            method: 请求方法 (get或post)
    """
    start_time = time.time()
    results = {
        "target": target,
        "vulnerability_type": VULNERABILITY_TYPE,
        "severity": SEVERITY,
        "scan_depth": kwargs.get("scan_depth", 2),
        "start_time": start_time,
        "end_time": None,
        "duration": None,
        "vulnerable": False,
        "vulnerabilities": [],
        "details": {
            "total_tests": 0,
            "passed_tests": 0,
            "failed_tests": 0,
            "test_details": []
        }
    }
    
    # 获取参数
    timeout = kwargs.get("timeout", 10)
    max_retries = kwargs.get("max_retries", 3)
    scan_depth = kwargs.get("scan_depth", 2)
    test_params = kwargs.get("test_params", [])
    method = kwargs.get("method", "get").lower()
    
    logger.info(f"开始不安全的反序列化漏洞扫描: {target} (方法: {method})")
    
    # 根据扫描深度选择要测试的序列化格式
    if scan_depth < 3:
        test_formats = SERIALIZATION_FORMATS[:2]  # 只测试前2种格式
    else:
        test_formats = SERIALIZATION_FORMATS
    
    # 1. 测试URL中的现有参数
    logger.info("正在测试URL中的现有参数")
    for format_info in test_formats:
        existing_params_test = test_url_parameters(target, format_info, method, timeout)
        results["details"]["test_details"].append({
            "test_type": "existing_params_test",
            "format": format_info["name"],
            "result": existing_params_test
        })
        results["details"]["total_tests"] += len(existing_params_test["results"])
        
        if existing_params_test["vulnerable"]:
            for result in existing_params_test["results"]:
                if result["vulnerable"]:
                    vulnerability = {
                        "url": target,
                        "param": result["param"],
                        "serialization_format": result["serialization_format"],
                        "language": result.get("language", "unknown"),
                        "error_indicators": result["error_indicators"],
                        "status_code": result["status_code"],
                        "base64_encoded": any(p["param"] == result["param"] for p in existing_params_test["base64_encoded_params"]),
                        "recommendation": "使用安全的序列化格式，实施输入验证，限制反序列化类的范围"
                    }
                    results["vulnerabilities"].append(vulnerability)
                    results["vulnerable"] = True
    
    # 2. 如果没有发现漏洞或扫描深度大于1，测试自定义参数
    if (not results["vulnerable"] or scan_depth > 1) and test_params:
        logger.info("正在测试自定义参数")
        for param_name in test_params:
            for format_info in test_formats[:2]:  # 为自定义参数测试前2种格式
                results["details"]["total_tests"] += 1
                
                try:
                    param_result = test_parameter_for_deserialization(
                        target, param_name, format_info, method, timeout
                    )
                    
                    if param_result["vulnerable"]:
                        vulnerability = {
                            "url": target,
                            "param": param_name,
                            "serialization_format": param_result["serialization_format"],
                            "language": param_result.get("language", "unknown"),
                            "error_indicators": param_result["error_indicators"],
                            "status_code": param_result["status_code"],
                            "recommendation": "使用安全的序列化格式，实施输入验证，限制反序列化类的范围"
                        }
                        results["vulnerabilities"].append(vulnerability)
                        results["vulnerable"] = True
                    
                    results["details"]["passed_tests"] += 1
                except Exception as e:
                    logger.error(f"测试参数 {param_name} 失败: {str(e)}")
                    results["details"]["failed_tests"] += 1
    
    # 3. 如果没有发现漏洞或扫描深度大于2，测试常见参数名
    if not results["vulnerable"] and scan_depth > 2:
        logger.info("正在测试常见参数名")
        for format_info in test_formats[:2]:  # 为常见参数测试前2种格式
            common_params_test = test_common_parameters(target, format_info, method, timeout)
            results["details"]["test_details"].append({
                "test_type": "common_params_test",
                "format": format_info["name"],
                "result": common_params_test
            })
            results["details"]["total_tests"] += len(common_params_test["results"])
            
            if common_params_test["vulnerable"]:
                for result in common_params_test["results"]:
                    if result["vulnerable"]:
                        vulnerability = {
                            "url": target,
                            "param": result["param"],
                            "serialization_format": result["serialization_format"],
                            "language": result.get("language", "unknown"),
                            "error_indicators": result["error_indicators"],
                            "status_code": result["status_code"],
                            "recommendation": "使用安全的序列化格式，实施输入验证，限制反序列化类的范围"
                        }
                        results["vulnerabilities"].append(vulnerability)
                        results["vulnerable"] = True
    
    # 计算结束时间和持续时间
    end_time = time.time()
    results["end_time"] = end_time
    results["duration"] = end_time - start_time
    
    # 生成总结报告
    if results["vulnerable"]:
        logger.warning(f"在 {target} 发现 {len(results['vulnerabilities'])} 个不安全的反序列化漏洞")
        results["summary"] = f"发现{len(results['vulnerabilities'])}个不安全的反序列化漏洞，这是严重的安全问题，攻击者可能远程执行代码或获取系统控制权"
    else:
        logger.info(f"{target} 未发现明显的不安全反序列化漏洞")
        results["summary"] = "未发现明显的不安全反序列化漏洞，但仍建议使用安全的序列化实践，如数据签名、类白名单和输入验证"
    
    return results